package cn.doitedu.api;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class _05_FileSource_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // 构建一个文件行格式的fileSource

        TextLineInputFormat textLineInputFormat = new TextLineInputFormat();
        FileSource<String> fileSource =
                FileSource
                        .forRecordStreamFormat(textLineInputFormat, new Path("data/text/"))
                        .monitorContinuously(Duration.ofSeconds(1))
                        .build();

        DataStreamSource<String> stream = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "file-source");

        stream.print();

        env.execute();




    }

}
